package com.gugee.ins.data.collector.blogger.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.spider.blogger.*;
import com.gugee.ins.data.collector.blogger.config.CollectorBloggerConfig;
import com.gugee.ins.data.collector.blogger.service.InsBloggerCollectService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.collector.blogger.service.InsBloggerFollowingCollectService;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerFollowingCollectServiceImpl implements InsBloggerFollowingCollectService {

    /**
     * 配置
     */
    @Autowired
    CollectorBloggerConfig collectorBloggerConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     *
     */
    @Autowired
    InsBloggerCollectService insBloggerCollectService;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insBloggerFeed-worker";


    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(collectorBloggerConfig.getInsBloggerFeedCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理insBlogger收入任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsBloggerFollowingCollect(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runCollectInsBloggerFollowing(msg);
                } catch (Exception ex) {
                    log.error("InsBloggerFollowingCollect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("InsBloggerFollowingCollect thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集播主关注人数据
     * @param insBloggerMessages
     */
    public void runCollectInsBloggerFollowing(InsBloggerMessage insBloggerMessages){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessages.getUid();
        // 根据uid爬取信息
        PageInfo pageInfo;

        pageInfo = spiderFollowing(uid,collectorBloggerConfig.getInsSpiderBloggerFollowingEachCount(),"");
        while(pageInfo != null && pageInfo.getHasNextPage() && !StringUtils.isEmpty(pageInfo.getEndCursor())){
            pageInfo = spiderFollowing(uid,collectorBloggerConfig.getInsSpiderBloggerFollowingEachCount(),pageInfo.getEndCursor());
        }

        if ((System.currentTimeMillis() - start) / 1000 > collectorBloggerConfig.getDebugTimeLimit()) {
            log.info("InsBlogger Collect uid:{} use time:{}s", insBloggerMessages.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 循环拉关注人
     * @param uid
     * @param count
     * @param endCursor
     * @return
     */
    private PageInfo spiderFollowing(long uid,int count,String endCursor){
        //根据uid爬取信息
        List<EdgeBlogger> edges;
        InsBloggerFollowingResponse insBloggerFollowingResponse;
        PageInfo pageInfo;
        try {
            insBloggerFollowingResponse = insBloggerSpiderClient.getFollowingBloggerByUid(uid,count,endCursor);
            edges = insBloggerFollowingResponse.getData().getUser().getEdgeFollow().getEdges();
            if(edges != null && !edges.isEmpty()){
                edges.forEach(edge -> {
                    if(edge != null && edge.getNode() != null
                            && !edge.getNode().getIsPrivate()
                            && edge.getNode().getUid() != null
                            && edge.getNode().getUid() > 0){
                        try{
                            insBloggerCollectService.sendCollectBloggerMsg(edge.getNode().getUid(),edge.getNode().getUsername(),edge.getNode().getPicUrl(),edge.getNode().getIsPrivate());
                        }catch (Exception e){
                            log.debug("spider ins blogger following get wrong uid");
                        }
                    }
                });
            }
            pageInfo = insBloggerFollowingResponse.getData().getUser().getEdgeFollow().getPageInfo();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins blogger following error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }



}
